package com.example.mq.mqserver.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 实现消费消息的核心
 */
public class ConsumerManager {
    //持有上层的 VirtualHost对象,便于操作数据.
    private VirtualHost parent;
    // 指定一个线程池, 负责去执行具体的回调任务.
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
    // 存放令牌的队列 --- 标志着当前哪个队列有消息
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    // 扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost p){
        parent = p;
        //创建扫描线程
        scannerThread = new Thread(()->{
            //扫描线程是一直运作的，进程不停它不停
            while (true){
                // 1.拿到有消息待消费的队列的队列名
                try {
                    String queueName = tokenQueue.take();
                    //2. 获取到该队列对象
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    //队列为空
                    if(queue == null){
                        throw new MqException("[ConsumerManager] 队列不存在! queueName=" + queueName);
                    }
                    //队列不为空 消费队列里的消息
                    synchronized (queue){
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    //调用时机:发送消息的时候 ---- 用于提醒消费者消费消息
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        // 找到对应的队列.
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在! queueName=" + queueName);
        }
        //创建消费者对象
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);
            // 如果当前队列中已经有了一些消息了, 需要立即就消费掉.
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                // 这个方法调用一次就消费一条消息.
                consumeMessage(queue);
            }
        }
    }

    private void consumeMessage(MSGQueue queue) {
        //通过轮询的方式 寻找一个消费者来消费消息
        ConsumerEnv consumer = queue.chooseConsumer();
        if(consumer == null){
            //当前队列没有消费者 暂时不消费,等到有消费者再消费
            return;
        }

        //获取一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message == null){
            //如果队列里面没有消息 也不进行消费
            return;
        }

        //把消息带入消费者的回调方法中,让线程池执行
        workerPool.submit(()->{
            try {
                //先将消息添加进 待确认消息集合 之中
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(),message);
                //消费者执行消费回调函数
                consumer.getConsumer().handleDelivery(consumer.getConsumerTag(),message.getBasicProperties(),
                        message.getBody());
                //如果是 自动应答 则直接删除消息 --- 内存 、硬盘、消息中心集合
                if(consumer.isAutoAck()){
                    // 1) 删除硬盘上的消息
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // 2) 删除上面的待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    // 3) 删除内存中消息中心里的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费. queueName=" + queue.getName());
                }
                //如果是 非自动应答 则执行 basicAck 的回调函数 主动删除消息
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}
